package lk.mrt.cse.wihidum.communicator;

import lk.mrt.cse.wihidum.processor.*;
import lk.mrt.cse.wihidum.storage.*;
import lk.mrt.cse.wihidum.util.*;
import org.apache.axiom.om.*;
import org.apache.axiom.soap.*;
import org.apache.axis2.*;
import org.apache.axis2.addressing.*;
import org.apache.axis2.client.*;
import org.apache.axis2.context.*;
import org.apache.axis2.wsdl.*;
import org.apache.log4j.*;

import javax.xml.namespace.QName;
import java.util.*;


public class ClusterRouter
        implements Router
{

    private static final Logger logger = Logger.getLogger(ClusterRouter.class);
    HashMap<String, String> myBrokerMap = SelfReader.read();

    private MessageTrace routerSubscriptionMessageTrace=new MessageTrace();

   // private BrokerSubscriptionDAO brokerSubscriptionDAO;

    public ClusterRouter()
    {
        //XMLReader xmlReader = new XMLReader("ClusterConfig.xml");
        if(!Topology.isReadOnce())
        {
            logger.info("Cluster Config was not read before.");
            XMLReader xmlReader = new XMLReader();
            //Use this object to get information needed.
            xmlReader.readXmlDocument("ClusterConfig.xml");
        }
    }

     public void setRouterSubscriptionMessageTrace(MessageTrace subscriptionMessageTrace)
     {
         
         this.routerSubscriptionMessageTrace.addToMessageTrace("topic",subscriptionMessageTrace.getFromMessageTrace("topic"));
         this.routerSubscriptionMessageTrace.addToMessageTrace("messageID",subscriptionMessageTrace.getFromMessageTrace("messageID"));
         this.routerSubscriptionMessageTrace.addToMessageTrace("action",subscriptionMessageTrace.getFromMessageTrace("action"));
         this.routerSubscriptionMessageTrace.addToMessageTrace("receivePath",subscriptionMessageTrace.getFromMessageTrace("receivePath"));

     }
    public void propagateSubscription(SOAPEnvelope message, String filter, String action,String logUUID)
    {
        SOAPEnvelope brokerSubscriptionEnvelope = null;
        //check my role
        String myRole = Topology.getMyRole();

        logger.info("This is a " + myRole);

       // brokerSubscriptionDAO = new BrokerSubscriptionDAO();

        //if i am a broker, cc or a scc
        if (myRole.equals("BROKER") || myRole.equals("CC") || myRole.equals("SCC"))
        {
            //get immediate supper unit
            NeighbourBroker superUnit = null;
            boolean alreadySubscribed = false;
            if (((Topology.getSuperUnits()).size()) != 0)//check whether immediate super units exist
            {
                superUnit = Topology.getSuperUnits().get(0);
                //checks whether the current filter is subscribed to that super unit
                //alreadySubscribed = brokerSubscriptionDAO.isSubscribedTo(superUnit, filter);
                alreadySubscribed=BrokerSubscriptionMap.isSubscribedTo(superUnit,filter);
            }

            //IFNOT save the broker subscription and put the header+distribution trace
            if ((superUnit != null) && (!alreadySubscribed))
            {
                //make a broker subscription object for the new broker subscription
                BrokerSubscription brokerSubscription = new BrokerSubscription();
                brokerSubscription.setToBrokerAddress(superUnit.getAddress());
                brokerSubscription.setToBrokerID(superUnit.getBrokerID());
                brokerSubscription.setTopic(filter);
                //add messageTracing
                String myID=myBrokerMap.get("id");
                String sendPath=myID+"-"+superUnit.getBrokerID();
                routerSubscriptionMessageTrace.addToMessageTrace("sendPath",sendPath);

                //see whether the message is a broker subscription or normal subscription
                //if normal subscription,
                if (action == WihidumConstants.Actions.Subscribe)
                {
                    //addBrokerSubscriptionHeader(message, superUnit, filter);
                    brokerSubscriptionEnvelope = createBrokerSubscriptionEnvelope(superUnit, filter,logUUID);
                }
                //if  broker subscripiton
                if (action == WihidumConstants.Operations.BROKER_SUBSCRIBE)
                {
                    //updateBrokerSubscriptionHeader(message,superUnit);
                    brokerSubscriptionEnvelope = updateBrokerSubscriptionEnvelope(message, superUnit, filter,logUUID);
                }
                //send the message and receive the acknowledge
                OMElement response = sendBrokerSubscription(brokerSubscriptionEnvelope, superUnit);

                if (response != null)//IF received: save broker subscription to broker subscriber
                {
                    //logger.info("broker subscription sent from broker " + Topology.getMyId() + " to broker " + superUnit.getBrokerID() + " successfully for the message");
                    logger.info("broker subscription sent successfully for the message#"+logUUID);
                    String brokerSubscriptionUUID=UUID.randomUUID().toString();
                    BrokerSubscriptionMap.saveSubscriber(brokerSubscriptionUUID,brokerSubscription);
                    //BrokerSubscriptionManager brokerSubsriptionDAO = new BrokerSubscriptionDAO();
                    //brokerSubsriptionDAO.saveBrokerSubscription(brokerSubscription,"Sent");

                }
                else //IF not received: get secondary super unit and make it immediate superunit and do the above procedure
                {
                    logger.info("Error in sending broker subscription from broker" + Topology.getMyId() + " to the super unit " + superUnit.getBrokerID());
                }
            }
            //if already subscribed to the super unit, ignores propagation
            if ((superUnit != null) && (alreadySubscribed))
            {
                //logger.info("Broker " + Topology.getMyId() + " is already subsbribed to its super unit broker " + superUnit.getBrokerID() + " on the topic " + filter+" for the message#"+logUUID);
                logger.info("Broker is already subsbribed to its super unit for the message#"+logUUID);

            }
            //if there is no super unit, ignores propagation
            if (superUnit == null)
            {
                logger.info("Broker doesn't have a super unit. Hence no upward propagation for the msg#"+logUUID);

            }

        }
        //or if i am a sscc
        if ((myRole.equals("SSCC")))
        {
            //get peers
            boolean alreadySubscribed = false;
            boolean equalFromANDPeerAddresses=false;
            ArrayList<NeighbourBroker> peerUnits = Topology.getPeerUnits();
            if ((peerUnits.size()) != 0)//check whether peer units exist
            {
                for (NeighbourBroker peerUnit : peerUnits)//for each peer unit
                {

                    //checks whether the current filter is subscribed to peer
                    //alreadySubscribed = brokerSubscriptionDAO.isSubscribedTo(peerUnit, filter);
                    alreadySubscribed=BrokerSubscriptionMap.isSubscribedTo(peerUnit,filter);

                    if ((!alreadySubscribed))
                    {
                        //make a broker subscription object for the new broker subscription
                        BrokerSubscription brokerSubscription = new BrokerSubscription();
                        brokerSubscription.setToBrokerAddress(peerUnit.getAddress());
                        brokerSubscription.setToBrokerID(peerUnit.getBrokerID());
                        brokerSubscription.setTopic(filter);

                        OMElement response=null;
                        //addBrokerSubscriptionHeader(message, peerUnit, filter);
                        if (action == WihidumConstants.Actions.Subscribe)
                        {
                            //addBrokerSubscriptionHeader(message, superUnit, filter);
                            brokerSubscriptionEnvelope = createBrokerSubscriptionEnvelope(peerUnit, filter,logUUID);
                            //send the message and receive the acknowledge
                            response = sendBrokerSubscription(brokerSubscriptionEnvelope, peerUnit);
                            //add messageTracing
                            String myID=myBrokerMap.get("id");
                            String sendPath=myID+"-"+peerUnit.getBrokerID();
                            routerSubscriptionMessageTrace.addToMessageTrace("sendPath",sendPath);

                        }
                        else if (action == WihidumConstants.Operations.BROKER_SUBSCRIBE)
                        {
                            //get from address
                            String fromBrokerAddress=message.getBody()
                                    .getFirstChildWithName(new QName(WSEventingConstants.WIHIDUM_NAMESPACE,"BrokerSubscription"))
                                    .getFirstChildWithName(new QName(WSEventingConstants.WIHIDUM_NAMESPACE,"From"))
                                    .getFirstChildWithName(new QName(WSEventingConstants.WIHIDUM_NAMESPACE,"Address"))
                                    .getText();
                            if(peerUnit.getAddress().equals(fromBrokerAddress))
                            {
                                    //logger.info("Peer Unit and From Broker are equal");
                                    equalFromANDPeerAddresses=true;
                            }
                            if(!equalFromANDPeerAddresses)
                            {
                                brokerSubscriptionEnvelope = updateBrokerSubscriptionEnvelope(message, peerUnit, filter,logUUID);
                                //send the message and receive the acknowledge
                                response = sendBrokerSubscription(brokerSubscriptionEnvelope, peerUnit);
                                //add messageTracing
                                String myID=myBrokerMap.get("id");
                                String sendPath=myID+"-"+peerUnit.getBrokerID();
                                routerSubscriptionMessageTrace.addToMessageTrace("sendPath",sendPath);
                            }
                            else
                            {
                                 logger.info("Broker subscription received from its peer unit for the message#"+logUUID);

                            }
                        }

                        if (response != null)//IF received: save broker subscription to broker subscriber
                        {
                            logger.info("Broker subscription is sent successfully#"+logUUID);
                            String brokerSubscriptionUUID=UUID.randomUUID().toString();
                            BrokerSubscriptionMap.saveSubscriber(brokerSubscriptionUUID,brokerSubscription);
                            //BrokerSubscriptionManager brokerSubsriptionDAO = new BrokerSubscriptionDAO();
                            //brokerSubsriptionDAO.saveBrokerSubscription(brokerSubscription,"Sent");

                            //sendMessageTrace();

                        }

                    }
                    //if already subscribed to the peer unit, ignores propagation
                    if (alreadySubscribed)
                    {
                       // logger.info("Broker " + Topology.getMyId() + " is already subsbribed to its peer unit broker " + peerUnit.getBrokerID() + " on the topic " + filter);
                        logger.info("Broker is already subsbribed to its peer unit for the message#"+logUUID);

                    }
                }
            }

            //if there is no peer unit, ignores propagation
            if (peerUnits.size() == 0)
            {
                logger.info("SSCC broker doesn't have a peer unit. Hence no propagation for the messsage#"+logUUID);

            }

        }
        //get the distribution trace tracked
        this.sendMessageTrace();

    }

    public void reRouteAllSubscriptions(String filter, String action,String logUUID)
      {
          SOAPEnvelope brokerSubscriptionEnvelope = null;
          //check my role
          String myRole = Topology.getMyRole();

          logger.info("This is a " + myRole);

         // brokerSubscriptionDAO = new BrokerSubscriptionDAO();
          action = WihidumConstants.Actions.Subscribe;
          Map<SubscriberIdentifier,Subscriber> subMap = SubscriptionMap.getAllSubscribers();
          if(!subMap.isEmpty())
          {
              for(Map.Entry<SubscriberIdentifier,Subscriber> entry:subMap.entrySet())
                {
                    filter = entry.getKey().getFilterTopic();
                    logUUID =  entry.getValue().getUuid();
                    
                }
          }
          //if i am a broker, cc or a scc
          if (myRole.equals("BROKER") || myRole.equals("CC") || myRole.equals("SCC"))
          {
              //get immediate supper unit
              NeighbourBroker superUnit = null;
              boolean alreadySubscribed = false;
              if (((Topology.getSuperUnits()).size()) != 0)//check whether immediate super units exist
              {
                  superUnit = Topology.getSuperUnits().get(0);
                  //checks whether the current filter is subscribed to that super unit
                  //alreadySubscribed = brokerSubscriptionDAO.isSubscribedTo(superUnit, filter);
                  alreadySubscribed=BrokerSubscriptionMap.isSubscribedTo(superUnit,filter);
              }

              //IFNOT save the broker subscription and put the header+distribution trace
              if ((superUnit != null) && (!alreadySubscribed))
              {
                  //make a broker subscription object for the new broker subscription
                  BrokerSubscription brokerSubscription = new BrokerSubscription();
                  brokerSubscription.setToBrokerAddress(superUnit.getAddress());
                  brokerSubscription.setToBrokerID(superUnit.getBrokerID());
                  brokerSubscription.setTopic(filter);
                  //add messageTracing
                  String myID=myBrokerMap.get("id");
                  String sendPath=myID+"-"+superUnit.getBrokerID();
                  routerSubscriptionMessageTrace.addToMessageTrace("sendPath",sendPath);

                  //see whether the message is a broker subscription or normal subscription
                  //if normal subscription,
                  if (action == WihidumConstants.Actions.Subscribe)
                  {
                      //addBrokerSubscriptionHeader(message, superUnit, filter);
                      brokerSubscriptionEnvelope = createBrokerSubscriptionEnvelope(superUnit, filter,logUUID);
                  }
                  //send the message and receive the acknowledge
                  OMElement response = sendBrokerSubscription(brokerSubscriptionEnvelope, superUnit);

                  if (response != null)//IF received: save broker subscription to broker subscriber
                  {
                      //logger.info("broker subscription sent from broker " + Topology.getMyId() + " to broker " + superUnit.getBrokerID() + " successfully for the message");
                      logger.info("broker subscription sent successfully for the message#"+logUUID);
                      String brokerSubscriptionUUID=UUID.randomUUID().toString();
                      BrokerSubscriptionMap.saveSubscriber(brokerSubscriptionUUID,brokerSubscription);
                      //BrokerSubscriptionManager brokerSubsriptionDAO = new BrokerSubscriptionDAO();
                      //brokerSubsriptionDAO.saveBrokerSubscription(brokerSubscription,"Sent");

                  }
                  else //IF not received: get secondary super unit and make it immediate superunit and do the above procedure
                  {
                      logger.info("Error in sending broker subscription from broker" + Topology.getMyId() + " to the super unit " + superUnit.getBrokerID());
                  }
              }
              //if already subscribed to the super unit, ignores propagation
              if ((superUnit != null) && (alreadySubscribed))
              {
                  //logger.info("Broker " + Topology.getMyId() + " is already subsbribed to its super unit broker " + superUnit.getBrokerID() + " on the topic " + filter+" for the message#"+logUUID);
                  logger.info("Broker is already subsbribed to its super unit for the message#"+logUUID);

              }
              //if there is no super unit, ignores propagation
              if (superUnit == null)
              {
                  logger.info("Broker doesn't have a super unit. Hence no upward propagation for the msg#"+logUUID);

              }

          }
          //or if i am a sscc
          if ((myRole.equals("SSCC")))
          {
              //get peers
              boolean alreadySubscribed = false;
              boolean equalFromANDPeerAddresses=false;
              ArrayList<NeighbourBroker> peerUnits = Topology.getPeerUnits();
              if ((peerUnits.size()) != 0)//check whether peer units exist
              {
                  for (NeighbourBroker peerUnit : peerUnits)//for each peer unit
                  {

                      //checks whether the current filter is subscribed to peer
                      //alreadySubscribed = brokerSubscriptionDAO.isSubscribedTo(peerUnit, filter);
                      alreadySubscribed=BrokerSubscriptionMap.isSubscribedTo(peerUnit,filter);

                      if ((!alreadySubscribed))
                      {
                          //make a broker subscription object for the new broker subscription
                          BrokerSubscription brokerSubscription = new BrokerSubscription();
                          brokerSubscription.setToBrokerAddress(peerUnit.getAddress());
                          brokerSubscription.setToBrokerID(peerUnit.getBrokerID());
                          brokerSubscription.setTopic(filter);

                          OMElement response=null;
                          //addBrokerSubscriptionHeader(message, peerUnit, filter);
                          if (action == WihidumConstants.Actions.Subscribe)
                          {
                              //addBrokerSubscriptionHeader(message, superUnit, filter);
                              brokerSubscriptionEnvelope = createBrokerSubscriptionEnvelope(peerUnit, filter,logUUID);
                              //send the message and receive the acknowledge
                              response = sendBrokerSubscription(brokerSubscriptionEnvelope, peerUnit);
                              //add messageTracing
                              String myID=myBrokerMap.get("id");
                              String sendPath=myID+"-"+peerUnit.getBrokerID();
                              routerSubscriptionMessageTrace.addToMessageTrace("sendPath",sendPath);

                          }
                          
                          if (response != null)//IF received: save broker subscription to broker subscriber
                          {
                              logger.info("Broker subscription is sent successfully#"+logUUID);
                              String brokerSubscriptionUUID=UUID.randomUUID().toString();
                              BrokerSubscriptionMap.saveSubscriber(brokerSubscriptionUUID,brokerSubscription);
                              //BrokerSubscriptionManager brokerSubsriptionDAO = new BrokerSubscriptionDAO();
                              //brokerSubsriptionDAO.saveBrokerSubscription(brokerSubscription,"Sent");

                              //sendMessageTrace();

                          }

                      }
                      //if already subscribed to the peer unit, ignores propagation
                      if (alreadySubscribed)
                      {
                         // logger.info("Broker " + Topology.getMyId() + " is already subsbribed to its peer unit broker " + peerUnit.getBrokerID() + " on the topic " + filter);
                          logger.info("Broker is already subsbribed to its peer unit for the message#"+logUUID);

                      }
                  }
              }

              //if there is no peer unit, ignores propagation
              if (peerUnits.size() == 0)
              {
                  logger.info("SSCC broker doesn't have a peer unit. Hence no propagation for the messsage#"+logUUID);

              }

          }
          //get the distribution trace tracked
          this.sendMessageTrace();

      }


  /*  public ArrayList<NeighbourBroker> getSuperUnits()
    {
        return topology.getSuperUnits();
    }

    public ArrayList<NeighbourBroker> getPeerUnits()
    {
        return topology.getPeerUnits();
    }

    public ArrayList<NeighbourBroker> getSubUnits()
    {
        return topology.getSubUnits();
    }

    public SOAPEnvelope createResponse()
    {
        return null;  //To change body of implemented methods use File | Settings | File Templates.
    }   */

    private SOAPEnvelope createBrokerSubscriptionEnvelope(NeighbourBroker neighbourBroker, String filter, String logUUID)
    {
        SOAPFactory soapFactory = OMAbstractFactory.getSOAP11Factory();
        SOAPEnvelope brokerSubscriptionEnvelope = soapFactory.getDefaultEnvelope();
        OMNamespace wsaNS = soapFactory.createOMNamespace(WSEventingConstants.ADDRESSING_NAMESPACE, WSEventingConstants.ADDRESSING_PREFIX);
        brokerSubscriptionEnvelope.declareNamespace(wsaNS);
        OMNamespace wihidumNS = soapFactory.createOMNamespace(WSEventingConstants.WIHIDUM_NAMESPACE, WSEventingConstants.WHIDUM_NAMESPACE_PREFIX);
        brokerSubscriptionEnvelope.declareNamespace(wihidumNS);

        SOAPBody soapBody = brokerSubscriptionEnvelope.getBody();
        OMElement brokerSubscriptionElement = soapFactory.createOMElement("BrokerSubscription", wihidumNS);
        OMElement actionElement = soapFactory.createOMElement("Action", wihidumNS);
        actionElement.setText(WihidumConstants.Operations.BROKER_SUBSCRIBE);
        brokerSubscriptionElement.addChild(actionElement);

        OMElement toElement = soapFactory.createOMElement("To", wihidumNS);
        OMElement toAdressElement = soapFactory.createOMElement("Address", wihidumNS);
        toAdressElement.setText(neighbourBroker.getAddress());
        toElement.addChild(toAdressElement);
        OMElement toIDElement=soapFactory.createOMElement("ID",wihidumNS);
        toIDElement.setText(neighbourBroker.getBrokerID());
        toElement.addChild(toIDElement);
        brokerSubscriptionElement.addChild(toElement);
        //get the broker ID and set that as reference properties
        //OMElement toReferencePropoerties=factory.createOMElement("ReferenceProperties",null);

        OMElement fromElement = soapFactory.createOMElement("From", wihidumNS);
        OMElement fromAddressElement = soapFactory.createOMElement("Address", wihidumNS);
        fromAddressElement.setText(myBrokerMap.get("address"));
        fromElement.addChild(fromAddressElement);
        OMElement fromIDElement=soapFactory.createOMElement("ID",wihidumNS);
        fromIDElement.setText(myBrokerMap.get("id"));
        fromElement.addChild(fromIDElement);
        brokerSubscriptionElement.addChild(fromElement);

        OMElement filterElement = soapFactory.createOMElement("Filter", wihidumNS);
        filterElement.setText(filter);
        brokerSubscriptionElement.addChild(filterElement);

        OMElement logIDElement=soapFactory.createOMElement("LogID",wihidumNS);
        logIDElement.setText(logUUID);
        brokerSubscriptionElement.addChild(logIDElement);

        soapBody.addChild(brokerSubscriptionElement);

        OMElement distributionTraceElement = soapFactory.createOMElement("DistributionTrace", wihidumNS);
        OMElement startStationElement = soapFactory.createOMElement("Stations", wihidumNS);
        startStationElement.setText(Topology.getMyId());
        distributionTraceElement.addChild(startStationElement);

        soapBody.addChild(distributionTraceElement);

        return brokerSubscriptionEnvelope;
    }

    private SOAPEnvelope updateBrokerSubscriptionEnvelope(SOAPEnvelope subscriptionMessage, NeighbourBroker nextBroker, String filter,String logID)
    {
        //logger.info("Updating the broker subscription");  //actually for the moment, it is creating a new envelope

        SOAPFactory factory = OMAbstractFactory.getSOAP11Factory();


        SOAPEnvelope brokerSubscriptionEnvelope = factory.getDefaultEnvelope();
        OMNamespace wsaNS = factory.createOMNamespace(WSEventingConstants.ADDRESSING_NAMESPACE, WSEventingConstants.ADDRESSING_PREFIX);
        brokerSubscriptionEnvelope.declareNamespace(wsaNS);
        OMNamespace wihidumNS = factory.createOMNamespace(WSEventingConstants.WIHIDUM_NAMESPACE, WSEventingConstants.WHIDUM_NAMESPACE_PREFIX);
        brokerSubscriptionEnvelope.declareNamespace(wihidumNS);

        SOAPBody soapBody = brokerSubscriptionEnvelope.getBody();
        OMElement brokerSubscriptionElement = factory.createOMElement("BrokerSubscription", wihidumNS);
        OMElement actionElement = factory.createOMElement("Action", wihidumNS);
        actionElement.setText(WihidumConstants.Operations.BROKER_SUBSCRIBE);
        brokerSubscriptionElement.addChild(actionElement);

        OMElement toElement = factory.createOMElement("To", wihidumNS);
        OMElement toAdressElement = factory.createOMElement("Address", wihidumNS);
        toAdressElement.setText(nextBroker.getAddress());
        toElement.addChild(toAdressElement);
        OMElement toIDElement=factory.createOMElement("ID",wihidumNS);
        toIDElement.setText(nextBroker.getBrokerID());
        toElement.addChild(toIDElement);

        brokerSubscriptionElement.addChild(toElement);

        OMElement fromElement = factory.createOMElement("From", wihidumNS);
        OMElement fromAddressElement = factory.createOMElement("Address", wihidumNS);
        fromAddressElement.setText(myBrokerMap.get("address"));
        fromElement.addChild(fromAddressElement);
        OMElement fromIDElement=factory.createOMElement("ID",wihidumNS);
        fromIDElement.setText(myBrokerMap.get("id"));
        fromElement.addChild(fromIDElement);

        brokerSubscriptionElement.addChild(fromElement);

        OMElement filterElement = factory.createOMElement("Filter", wihidumNS);
        filterElement.setText(filter);
        brokerSubscriptionElement.addChild(filterElement);

        OMElement logIDElement=factory.createOMElement("LogID",wihidumNS);
        logIDElement.setText(logID);
        brokerSubscriptionElement.addChild(logIDElement);

        soapBody.addChild(brokerSubscriptionElement);

        return brokerSubscriptionEnvelope;


    }

    private OMElement sendBrokerSubscription(SOAPEnvelope brokerSubscriptionMessage, NeighbourBroker neighbourBroker)
    {

        OMElement responseElement = null;
        try
        {
            ServiceClient serviceClient = new ServiceClient(ClientConfiguration.getConfigurationContext(),null);
            //creating in out operating client
            OperationClient opClient = serviceClient.createClient(ServiceClient.ANON_OUT_IN_OP);
            //creating message context
            MessageContext outMsgCtx = new MessageContext();
            Options opts = outMsgCtx.getOptions();
            //setting properties into option
            opts.setTo(new EndpointReference(neighbourBroker.getAddress()));
            opts.setAction(WihidumConstants.Operations.BROKER_SUBSCRIBE);

            outMsgCtx.setEnvelope(brokerSubscriptionMessage);


            opClient.addMessageContext(outMsgCtx);
            opClient.execute(true);

            MessageContext inMsgtCtx = opClient.getMessageContext(WSDLConstants.MESSAGE_LABEL_IN_VALUE);

            SOAPEnvelope response = inMsgtCtx.getEnvelope();
            responseElement = response.getBody().getFirstElement();
        }
        catch (Exception ex)
        {
            ex.printStackTrace();
        }
        return responseElement;
    }

    private void sendMessageTrace()
    {
        try
                {
                    logger.info ("Sending subscription trace message");

                    String adminAddress=myBrokerMap.get("admin");
                    ServiceClient serviceClient = ClientConfiguration.getServiceClient();

                    //creating in out operating client
                    OperationClient opClient = serviceClient.createClient(ServiceClient.ANON_OUT_IN_OP);

                    //creating message context
                    MessageContext outMsgCtx = new MessageContext();
                    Options opts = outMsgCtx.getOptions();
                    //setting properties into option
                    //opts.setTo(clientbean.getBrokerAddress ());
                    opts.setTo(new EndpointReference(adminAddress));

                    opts.setAction(WihidumConstants.Operations.SET_MESSAGE_TRACE);

                    //serviceClient.engageModule("addressing");
                    String SOAPVersion = opts.getSoapVersionURI();
                    if (SOAPVersion==null)
                        SOAPVersion=SOAP12Constants.SOAP_ENVELOPE_NAMESPACE_URI;
                    SOAPEnvelope subscriptionTraceEnvelope=createTraceMessageEnvelope();
                    //after making the traceEnvelope, clear the MessageTrace data.
                    routerSubscriptionMessageTrace.clearMessageTrace();

                    outMsgCtx.setEnvelope(subscriptionTraceEnvelope);
                    opClient.addMessageContext(outMsgCtx);
                    opClient.execute(true);

                    logger.info("Subscription Trace Message sent to: "+adminAddress);
                }
                catch(Exception e)
                {        e.printStackTrace();

                }

    }

    private SOAPEnvelope createTraceMessageEnvelope()
    {
        SOAPFactory factory=OMAbstractFactory.getSOAP12Factory();

        SOAPEnvelope soapEnvelope=factory.getDefaultEnvelope();
        //declare the used namespaces in the SOAP Envelope

        OMNamespace wsaNS=factory.createOMNamespace(WSEventingConstants.ADDRESSING_NAMESPACE,WSEventingConstants.ADDRESSING_PREFIX);
        soapEnvelope.declareNamespace(wsaNS);
        OMNamespace wihidumNS=factory.createOMNamespace(WSEventingConstants.WIHIDUM_NAMESPACE,WSEventingConstants.WHIDUM_NAMESPACE_PREFIX);
        soapEnvelope.declareNamespace(wihidumNS);

        SOAPBody soapBody=soapEnvelope.getBody();

        //create the trace element for the body
        OMElement traceElement =factory.createOMElement(WSEventingConstants.ElementNames.DistributionTrace, wihidumNS);
        String distributionTraceText= routerSubscriptionMessageTrace.getFromMessageTrace("messageID")+
                "#"+routerSubscriptionMessageTrace.getFromMessageTrace("action")+
                "#"+routerSubscriptionMessageTrace.getFromMessageTrace("topic")+
                "#"+routerSubscriptionMessageTrace.getFromMessageTrace("receivePath");
        if(routerSubscriptionMessageTrace.getFromMessageTrace("sendPath") != null)
        {
               distributionTraceText = distributionTraceText+","+routerSubscriptionMessageTrace.getFromMessageTrace("sendPath");
        }
        traceElement.setText(distributionTraceText);
        logger.info("Subscription trace sent from broker: "+distributionTraceText);
        soapBody.addChild(traceElement);

         return soapEnvelope;
    }
    public String propagatePublication(PublicationMessage message)
    {
        //check my role
        String myRole = Topology.getMyRole();
        logger.info(myRole);

        //if i am a broker, cc or a scc
        if (myRole.equals("BROKER") || myRole.equals("CC") || myRole.equals("SCC"))
        {
            //get immediate supper unit
            NeighbourBroker superUnit = null;
            if (((Topology.getSuperUnits()).size()) != 0)//check whether immediate super units exist
            {
                superUnit = Topology.getSuperUnits().get(0);
            }

            if (superUnit != null)
            {
                if (!superUnit.getAddress().equals(message.getBrokerPublicationElementFrom()))
                {                   
                    OMElement response = this.sendBrokerPublication(message, superUnit);
                    if (response != null)//IF received: save broker publication to broker subscriber
                    {
                        logger.info("Broker publication sent from broker " + Topology.getMyId() + " to broker " + superUnit.getBrokerID() + " successfully");
                        return superUnit.getBrokerID();
                    }
                    else //IF not received: get secondary super unit and make it immediate superunit and do the above procedure
                    {
                        logger.debug("Error in sending broker publication from broker" + Topology.getMyId() + " to the super unit " + superUnit.getBrokerID());
                        return null;
                    }
                }
                else
                {
                    logger.info("Broker publication received from the super unit " + superUnit.getBrokerID() + ". Hence, no need to route upwards.");
                    return null;
                }

            }
            //if there is no super unit, ignores propagation
            else
            {
                logger.debug("Broker " + Topology.getMyId() + " doesn't have a super unit. Hence no upward propagation after this.");
                return null;
            }
        }
        else
             return null;
    }

    private OMElement addBrokerPublicationElement(NeighbourBroker neighbourBroker, OMNamespace wihidumNS, String messageID)
       {
           logger.debug("Adding the broker publication element");
           SOAPFactory factory = OMAbstractFactory.getSOAP11Factory();

           OMElement brokerPublicationElement = factory.createOMElement(WihidumConstants.MessageElements.BROKER_PUBLICATION_HEADER, wihidumNS);
           OMElement actionElement = factory.createOMElement("Action", wihidumNS);
           actionElement.setText(WihidumConstants.Operations.BROKER_PUBLISH);
           brokerPublicationElement.addChild(actionElement);

           OMElement messageIDElement = factory.createOMElement("MessageID", wihidumNS);
           messageIDElement.setText(messageID);
           brokerPublicationElement.addChild(messageIDElement);

           OMElement toElement = factory.createOMElement("To", wihidumNS);
           OMElement toAdressElement = factory.createOMElement("Address", wihidumNS);
           toAdressElement.setText(neighbourBroker.getAddress());
           toElement.addChild(toAdressElement);
           brokerPublicationElement.addChild(toElement);

           OMElement fromElement = factory.createOMElement("From", wihidumNS);
           OMElement fromAddressElement = factory.createOMElement("Address", wihidumNS);
           fromAddressElement.setText(Topology.getMyAddress());
           logger.info("Setting my address"+fromAddressElement.getText()+"as the from broker in broker publication for the first time");
           fromElement.addChild(fromAddressElement);
           brokerPublicationElement.addChild(fromElement);

           OMElement distributionTraceElement = factory.createOMElement("DistributionTrace", wihidumNS);
           OMElement startStationElement = factory.createOMElement("StartStation", wihidumNS);
           startStationElement.setText(Topology.getMyId());
           distributionTraceElement.addChild(startStationElement);
           brokerPublicationElement.addChild(distributionTraceElement);

           return brokerPublicationElement;
       }

       private OMElement updateBrokerPublicationElement(PublicationMessage publicationMessage, NeighbourBroker neighbourBroker, OMNamespace wihidumNS)
       {
           logger.debug("Updating the broker publication element");

           SOAPFactory factory = OMAbstractFactory.getSOAP11Factory();

           OMElement brokerPublicationElement = factory.createOMElement("BrokerPublicationHeader", wihidumNS);
           OMElement actionElement = factory.createOMElement("Action", wihidumNS);
           actionElement.setText(WihidumConstants.Operations.BROKER_PUBLISH);
           brokerPublicationElement.addChild(actionElement);

           OMElement messageIDElement = factory.createOMElement("MessageID", wihidumNS);
           messageIDElement.setText(publicationMessage.getMessageID());
           brokerPublicationElement.addChild(messageIDElement);

           OMElement toElement = factory.createOMElement("To", wihidumNS);
           OMElement toAdressElement = factory.createOMElement("Address", wihidumNS);
           toAdressElement.setText(neighbourBroker.getAddress());
           toElement.addChild(toAdressElement);
           brokerPublicationElement.addChild(toElement);

           OMElement fromElement = factory.createOMElement("From", wihidumNS);
           OMElement fromAddressElement = factory.createOMElement("Address", wihidumNS);
           fromAddressElement.setText(Topology.getMyAddress());
           logger.info("Setting my address"+fromAddressElement.getText()+"as the from broker in broker publication");
           fromElement.addChild(fromAddressElement);
           brokerPublicationElement.addChild(fromElement);

           OMElement distributionTraceElement = factory.createOMElement("DistributionTrace", wihidumNS);
           OMElement startStationElement = factory.createOMElement("StartStation", wihidumNS);
           startStationElement.setText(publicationMessage.getBrokerPublicationElementStartStation());
           distributionTraceElement.addChild(startStationElement);

           ArrayList<String> intermediateStations = publicationMessage.getBrokerPublicationElementIntermediateStations();

           for (String station : intermediateStations)
           {
               OMElement intermediateStationElement = factory.createOMElement("IntermediateStation", wihidumNS);
               intermediateStationElement.setText(station);
               distributionTraceElement.addChild(intermediateStationElement);
           }
           OMElement intermediateStationElement = factory.createOMElement("IntermediateStation", wihidumNS);
           intermediateStationElement.setText(Topology.getMyId());
           distributionTraceElement.addChild(intermediateStationElement);
           brokerPublicationElement.addChild(distributionTraceElement);

           return brokerPublicationElement;
       }

       private OMElement sendBrokerPublication(PublicationMessage brokerPublicationMessage, NeighbourBroker neighbourBroker)
       {
           logger.debug("Sending publication message");
           OMElement responseElement = null;
           ServiceClient serviceClient = null;

           try
           {
               serviceClient = new ServiceClient();

               //creating in out operating client
               OperationClient opClient = serviceClient.createClient(ServiceClient.ANON_OUT_IN_OP);
               //creating message context
               MessageContext outMsgCtx = new MessageContext();
               Options opts = outMsgCtx.getOptions();
               //setting properties into option
               opts.setTo(new EndpointReference(neighbourBroker.getAddress()));
               opts.setAction(WihidumConstants.Operations.BROKER_PUBLISH);

               String SOAPVersion = opts.getSoapVersionURI();
               if (SOAPVersion == null)
               {
                   SOAPVersion = SOAP12Constants.SOAP_ENVELOPE_NAMESPACE_URI;
               }

               SOAPEnvelope brokerPublicationEnvelope = this.createPublicationRequest(brokerPublicationMessage, neighbourBroker, SOAPVersion);

               outMsgCtx.setEnvelope(brokerPublicationEnvelope);
               opClient.addMessageContext(outMsgCtx);
               opClient.execute(true);

               MessageContext inMsgtCtx = opClient.getMessageContext(WSDLConstants.MESSAGE_LABEL_IN_VALUE);

               SOAPEnvelope response = inMsgtCtx.getEnvelope();
               responseElement = response.getBody().getFirstElement();
               logger.info("Event#"+ brokerPublicationMessage.getMessageID() +" was routed successfully");
           }
           catch (AxisFault axisFault)
           {
               logger.error(axisFault);
           }
           return responseElement;
       }

       public SOAPEnvelope createPublicationRequest(PublicationMessage publicationMessage, NeighbourBroker superUnit, String SOAPURI)
       {
           try
           {
               SOAPFactory soapFactory = null;
               if (SOAP11Constants.SOAP_ENVELOPE_NAMESPACE_URI.equals(SOAPURI))
               {
                   soapFactory = OMAbstractFactory.getSOAP11Factory();
               }
               else
               {
                   if (SOAP12Constants.SOAP_ENVELOPE_NAMESPACE_URI.equals(SOAPURI))
                   {
                       soapFactory = OMAbstractFactory.getSOAP12Factory();
                   }
                   else
                   {
                       throw new Exception("Unknown SOAP version");
                   }
               }

               SOAPEnvelope soapEnvelope = soapFactory.getDefaultEnvelope();

               //declare the used namespaces in the SOAP Envelope
               OMNamespace wseventingNS = soapFactory.createOMNamespace(WSEventingConstants.EVENTING_NAMESPACE, WSEventingConstants.EVENTING_PREFIX);
               soapEnvelope.declareNamespace(wseventingNS);
               OMNamespace wsaNS = soapFactory.createOMNamespace(WSEventingConstants.ADDRESSING_NAMESPACE, WSEventingConstants.ADDRESSING_PREFIX);
               soapEnvelope.declareNamespace(wsaNS);
               OMNamespace wihidumNS = soapFactory.createOMNamespace(WSEventingConstants.WIHIDUM_NAMESPACE, WSEventingConstants.WHIDUM_NAMESPACE_PREFIX);
               soapEnvelope.declareNamespace(wihidumNS);

               OMFactory factory = OMAbstractFactory.getOMFactory();

               SOAPHeader soapHeader = soapEnvelope.getHeader();


               OMElement publicationHeader = factory.createOMElement(WihidumConstants.MessageElements.PUBLICATION_HEADER, wihidumNS);

               OMElement replyTo = factory.createOMElement("ReplyTo", wihidumNS);
               OMElement address = factory.createOMElement("Address", wihidumNS);
               address.setText(publicationMessage.getEventSourceURL());
               replyTo.addChild(address);
               publicationHeader.addChild(replyTo);

               OMElement topicElement = factory.createOMElement("Topic", wihidumNS);
               topicElement.setText(publicationMessage.getTopic());
               publicationHeader.addChild(topicElement);

               soapHeader.addChild(publicationHeader);

               SOAPBody soapBody = soapEnvelope.getBody();
               soapBody.setText(publicationMessage.getBody());

               OMElement brokerPublicationHeader = null;
               if (publicationMessage.getAction().equals(WihidumConstants.Operations.BROKER_PUBLISH))
               {
                   brokerPublicationHeader = this.updateBrokerPublicationElement(publicationMessage, superUnit, wihidumNS);
               }
               else
               {
                   brokerPublicationHeader = this.addBrokerPublicationElement(superUnit, wihidumNS, publicationMessage.getMessageID());
               }

               soapHeader.addChild(brokerPublicationHeader);

               return soapEnvelope;
           }
           catch (Exception ex)
           {
               logger.error(ex);
               return null;
           }

       }

}
